package com.jianger.service;

import com.alibaba.fastjson.JSONObject;
import com.jianger.conf.ConfigurationManager;
import com.jianger.conf.Constants;
import com.jianger.conf.DaoFactory;
import com.jianger.conf.MockData;
import com.jianger.dao.ISessionAggrStatDAO;
import com.jianger.dao.ISessionDeatilDao;
import com.jianger.dao.ISessionRandomExtractDAO;
import com.jianger.dao.TaskDao;
import com.jianger.pojo.SessionAggrStat;
import com.jianger.pojo.SessionRandomEx;
import com.jianger.pojo.Sessiondetail;
import com.jianger.pojo.Task;
import com.jianger.utils.*;
import org.apache.spark.Accumulator;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.hive.HiveContext;
import scala.Tuple2;

import java.sql.SQLException;
import java.util.*;

/**
 * 用户访问session分析Spark作业
 *
 * @author Administrator
 */
public class UserVisitSessionAnalyzeSpark {

    public static void main(String[] args) {

        args = new String[]{"2"};

        //1. 构建Spark上下文
        SparkConf conf = new SparkConf()
                .setAppName(Constants.SPARK_CONF_APPNAME)
                .setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);
        SQLContext sqlContext = getSQLContext(sc.sc());
        //2. 生成模拟测试数据
        mockData(sc, sqlContext);

        //3. 查询task任务
        TaskDao taskDao = DaoFactory.getTaskDaoImpl();
        long task_id = ParamUtils.getTaskIdFromArgs(args);

        // 4. 根据taskid查询task，并获取任务参数
        Task task = taskDao.findTaskById(task_id);
        System.out.println(task.getTask_param());
        JSONObject params = JSONObject.parseObject(task.getTask_param());
        // 5. 根据时间参数，获取sessionjavaRdd

        /**
         * (row)--->每行的数据
         */
        JavaRDD<Row> actionRdd = getActionRDDByDateRange(sqlContext, params);
        /**
         *  //（sessionId,row）---->sessionid加每行的数据
         */
        JavaPairRDD<String, Row> session2action = getSessionid2ActionRDD(actionRdd);

        // 6. session为key的rdd
        // 首先，可以将行为数据，按照session_id进行groupByKey分组
        // 此时的数据的粒度就是session粒度了，然后呢，可以将session粒度的数据
        // 与用户信息数据，进行join
        // 然后就可以获取到session粒度的数据，同时呢，数据里面还包含了session对应的user的信息
        JavaPairRDD<String, String> sessionid2AggrInfoRDD = aggregateBySession(sqlContext, actionRdd, session2action);


        // 根据各种行为参数过滤

        // 接着，就要针对session粒度的聚合数据，按照使用者指定的筛选参数进行数据过滤
        // 相当于我们自己编写的算子，是要访问外面的任务参数对象的
        // 所以，大家记得我们之前说的，匿名内部类（算子函数），访问外部对象，是要给外部对象使用final修饰的
        // 重构，同时进行过滤和统计
        Accumulator<String> sessionAggrStatAccumulator = sc.accumulator(
                "", new SessionAggrStatAccumulator());

        JavaPairRDD<String, String> filteredSessionid2AggrInfoRDD = filterSessionAndAggrStat(
                sessionid2AggrInfoRDD, params, sessionAggrStatAccumulator);
        filteredSessionid2AggrInfoRDD.count();


        // 随机选取session
        randomExtractSession(task.getTask_id(), filteredSessionid2AggrInfoRDD, session2action);

        /**
         * 特别说明
         * 我们知道，要将上一个功能的session聚合统计数据获取到，就必须是在一个action操作触发job之后
         * 才能从Accumulator中获取数据，否则是获取不到数据的，因为没有job执行，Accumulator的值为空
         * 所以，我们在这里，将随机抽取的功能的实现代码，放在session聚合统计功能的最终计算和写库之前
         * 因为随机抽取功能中，有一个countByKey算子，是action操作，会触发job
         */

        // 计算出各个范围的session占比，并写入MySQL
        calculateAndPersistAggrStat(sessionAggrStatAccumulator.value(),
                task.getTask_id());
        // 关闭Spark上下文
        sc.close();
    }


    private static void randomExtractSession(final long task_id, JavaPairRDD<String, String> filteredSessionid2AggrInfoRDD, JavaPairRDD<String, Row> session2action) {
        /**
         * 第一步，计算出每天每小时的session数量
         */

        // 获取<yyyy-MM-dd_HH,aggrInfo>格式的RDD

        JavaPairRDD<String, String> timeSession = filteredSessionid2AggrInfoRDD.mapToPair(new PairFunction<Tuple2<String, String>, String, String>() {
            @Override
            public Tuple2<String, String> call(Tuple2<String, String> tuple2) throws Exception {

                // 获取聚合数据
                String aggInfo = tuple2._2;
                // 获取start_time
                String startTime = StringUtils.getFieldFromConcatString(aggInfo, "\\|", Constants.FIELD_START_TIME);
                // 获取小时
                String hour = DateUtils.getDateHour(startTime);
                return new Tuple2<String, String>(hour, aggInfo);
            }
        });

        /**
         * 思考一下：这里我们不要着急写大量的代码，做项目的时候，一定要用脑子多思考
         *
         * 每天每小时的session数量，然后计算出每天每小时的session抽取索引，遍历每天每小时session
         * 首先抽取出的session的聚合数据，写入session_random_extract表
         * 所以第一个RDD的value，应该是session聚合数据
         *
         */

        // 得到每天每小时的session数量
        /**
         * 2018-04-06_02,27
         * (每天每小时需要的数量)
         */
        Map<String, Object> countMap = timeSession.countByKey();


        /**
         * 第二步，使用按时间比例随机抽取算法，计算出每天每小时要抽取session的索引
         */

        // 将<yyyy-MM-dd_HH,count>格式的map，转换成<yyyy-MM-dd,<HH,count>>的格式
        Map<String, Map<String, Long>> dateHourMap = new HashMap<String, Map<String, Long>>();



        // 遍历countMap,获取相关值

        for (Map.Entry<String, Object> countEntry : countMap.entrySet()) {
            // 获取日期
            String key = countEntry.getKey();
            String date = key.split("_")[0];
            String hour = key.split("_")[1];
            Long count = (Long) countEntry.getValue();
            //把日期，小时，和count放进去
            Map<String, Long> hourCount = dateHourMap.get(date);
            if (hourCount == null) {
                hourCount = new HashMap<String, Long>();
                dateHourMap.put(date, hourCount);
            }
            hourCount.put(hour, count);
        }

        // 开始实现我们的按时间比例随机抽取算法

        // 总共要抽取100个session，先按照天数，进行平分
        int extractNumberPerDay = 100 / dateHourMap.size();

        // <date,<hour,(3,5,20,102)>>
        final Map<String, Map<String, List<Integer>>> dateHourExtractMap =
                new HashMap<String, Map<String, List<Integer>>>();

        Random random = new Random();

        for (Map.Entry<String, Map<String, Long>> dateHourCountEntry : dateHourMap.entrySet()) {
            String date = dateHourCountEntry.getKey();
            Map<String, Long> hourCountMap = dateHourCountEntry.getValue();

            // 计算出这一天的session总数
            long sessionCount = 0L;
            for (long hourCount : hourCountMap.values()) {
                sessionCount += hourCount;
            }

            Map<String, List<Integer>> hourExtractMap = dateHourExtractMap.get(date);
            if (hourExtractMap == null) {
                hourExtractMap = new HashMap<String, List<Integer>>();
                dateHourExtractMap.put(date, hourExtractMap);
            }

            // 遍历每个小时
            for (Map.Entry<String, Long> hourCountEntry : hourCountMap.entrySet()) {
                String hour = hourCountEntry.getKey();
                long count = hourCountEntry.getValue();

                // 计算每个小时的session数量，占据当天总session数量的比例，直接乘以每天要抽取的数量
                // 就可以计算出，当前小时需要抽取的session数量
                int hourExtractNumber = (int) (((double) count / (double) sessionCount)
                        * extractNumberPerDay);
                if (hourExtractNumber > count) {
                    hourExtractNumber = (int) count;
                }

                // 先获取当前小时的存放随机数的list
                List<Integer> extractIndexList = hourExtractMap.get(hour);
                if (extractIndexList == null) {
                    extractIndexList = new ArrayList<Integer>();
                    hourExtractMap.put(hour, extractIndexList);
                }

                // 生成上面计算出来的数量的随机数
                for (int i = 0; i < hourExtractNumber; i++) {
                    int extractIndex = random.nextInt((int) count);
                    while (extractIndexList.contains(extractIndex)) {
                        extractIndex = random.nextInt((int) count);
                    }
                    extractIndexList.add(extractIndex);
                }
            }
        }

        for (Map.Entry<String, Map<String, List<Integer>>> datehour : dateHourExtractMap.entrySet()) {

            String key = datehour.getKey();
            Map<String, List<Integer>> value = datehour.getValue();
            for (Map.Entry<String, List<Integer>> stringListEntry : value.entrySet()) {
                String key1 = stringListEntry.getKey();
                List<Integer> value1 = stringListEntry.getValue();
                for (Integer integer : value1) {
                    System.out.println(integer);
                }

            }


        }



        /**
         * 第三步：遍历每天每小时的session，然后根据随机索引进行抽取
         */

        // 执行groupByKey算子，得到<dateHour,(session aggrInfo)>
        JavaPairRDD<String, Iterable<String>> time2sessionsRDD = timeSession.groupByKey();

        // 我们用flatMap算子，遍历所有的<dateHour,(session aggrInfo)>格式的数据
        // 然后呢，会遍历每天每小时的session
        // 如果发现某个session恰巧在我们指定的这天这小时的随机抽取索引上
        // 那么抽取该session，直接写入MySQL的random_extract_session表
        // 将抽取出来的session id返回回来，形成一个新的JavaRDD<String>
        // 然后最后一步，是用抽取出来的sessionid，去join它们的访问行为明细数据，写入session表
        JavaPairRDD<String, String> extractSessionidsRDD = time2sessionsRDD.flatMapToPair(

                new PairFlatMapFunction<Tuple2<String, Iterable<String>>, String, String>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public Iterable<Tuple2<String, String>> call(
                            Tuple2<String, Iterable<String>> tuple)
                            throws Exception {
                        List<Tuple2<String, String>> extractSessionids =
                                new ArrayList<Tuple2<String, String>>();

                        String dateHour = tuple._1;
                        String date = dateHour.split("_")[0];
                        String hour = dateHour.split("_")[1];
                        Iterator<String> iterator = tuple._2.iterator();

                        //每天每小时的session
                        List<Integer> extractIndexList = dateHourExtractMap.get(date).get(hour);
                        int size = dateHourExtractMap.size();
                        System.out.println(size);

                        ISessionRandomExtractDAO sessionRandomExtractDAO =
                                DaoFactory.getSessionExtractDaoImpl();


                        // 遍历每天每小时的session
                        int index = 0;
                        while (iterator.hasNext()) {
                            String sessionAggrInfo = iterator.next();

                            if (extractIndexList.contains(index)) {
                                String sessionid = StringUtils.getFieldFromConcatString(
                                        sessionAggrInfo, "\\|", Constants.FIELD_SESSION_ID);

                                // 将数据写入MySQL
                                SessionRandomEx sessionRandomExtract = new SessionRandomEx();
                                sessionRandomExtract.setTask_id(task_id);
                                sessionRandomExtract.setSession_id(sessionid);
                                sessionRandomExtract.setStart_time(StringUtils.getFieldFromConcatString(
                                        sessionAggrInfo, "\\|", Constants.FIELD_START_TIME));
                                sessionRandomExtract.setSearch_keywords(StringUtils.getFieldFromConcatString(
                                        sessionAggrInfo, "\\|", Constants.FIELD_SEARCH_KEYWORDS));
                                sessionRandomExtract.setClick_categoryids(StringUtils.getFieldFromConcatString(
                                        sessionAggrInfo, "\\|", Constants.FIELD_CLICK_CATEGORY_IDS));

                                sessionRandomExtract.setEnd_time(StringUtils.getFieldFromConcatString(sessionAggrInfo, "\\|", Constants.FIELD_END_TIME));

                                sessionRandomExtractDAO.save(sessionRandomExtract);

                                // 将sessionid加入list
                                extractSessionids.add(new Tuple2<String, String>(sessionid, sessionid));
                            }

                            index++;
                        }

                        return extractSessionids;
                    }

                });


        /**
         * 第四步，获取随机抽取的session的详细数据
         */

        // 1.把seesion之前的跟聚合的拼接起来
        JavaPairRDD<String, Tuple2<String, Row>> extractSession2ActioRdd = extractSessionidsRDD.join(session2action);
        extractSession2ActioRdd.foreach(new VoidFunction<Tuple2<String, Tuple2<String, Row>>>() {
            @Override
            public void call(Tuple2<String, Tuple2<String, Row>> tuple2) throws Exception {

                // 获取join的sesion详细数据
                Row row = tuple2._2._2;

                // 封装sessionDetail

                Sessiondetail sessiondetail = new Sessiondetail();
                sessiondetail.setTask_id(task_id);
                sessiondetail.setUser_id(row.getLong(1));
                sessiondetail.setSession_id(row.getString(2));
                sessiondetail.setPage_id(row.getLong(3));
                sessiondetail.setAction_time(row.getString(4));
                sessiondetail.setSearch_keyword(row.getString(5));
                sessiondetail.setClick_category_id(row.getLong(6));
                sessiondetail.setClick_product_id(row.getLong(7));
                sessiondetail.setOrder_category_ids(row.getString(8));
                sessiondetail.setOrder_product_ids(row.getString(9));
                sessiondetail.setPay_category_ids(row.getString(10));
                sessiondetail.setPay_product_ids(row.getString(11));
                // 保存到数据库
                ISessionDeatilDao iSessionDeatilDao = DaoFactory.getSessionDetailImpl();
                iSessionDeatilDao.save(sessiondetail);
            }
        });


    }

    private static void calculateAndPersistAggrStat(String value, Long task_id) {
        System.out.println(value);
        /**
         * 计算各session范围占比，并写入MySQL
         * @param value
         */
        // 从Accumulator统计串中获取值
        long session_count = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.SESSION_COUNT));

        long visit_length_1s_3s = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_1s_3s));
        long visit_length_4s_6s = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_4s_6s));
        long visit_length_7s_9s = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_7s_9s));
        long visit_length_10s_30s = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_10s_30s));
        long visit_length_30s_60s = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_30s_60s));
        long visit_length_1m_3m = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_1m_3m));
        long visit_length_3m_10m = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_3m_10m));
        long visit_length_10m_30m = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_10m_30m));
        long visit_length_30m = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_30m));

        long step_length_1_3 = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.STEP_PERIOD_1_3));
        long step_length_4_6 = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.STEP_PERIOD_4_6));
        long step_length_7_9 = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.STEP_PERIOD_7_9));
        long step_length_10_30 = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.STEP_PERIOD_10_30));
        long step_length_30_60 = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.STEP_PERIOD_30_60));
        long step_length_60 = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.STEP_PERIOD_60));

        // 计算各个访问时长和访问步长的范围
        double visit_length_1s_3s_ratio = NumberUtils.formatDouble(
                (double) visit_length_1s_3s / (double) session_count, 2);
        double visit_length_4s_6s_ratio = NumberUtils.formatDouble(
                (double) visit_length_4s_6s / (double) session_count, 2);
        double visit_length_7s_9s_ratio = NumberUtils.formatDouble(
                (double) visit_length_7s_9s / (double) session_count, 2);
        double visit_length_10s_30s_ratio = NumberUtils.formatDouble(
                (double) visit_length_10s_30s / (double) session_count, 2);
        double visit_length_30s_60s_ratio = NumberUtils.formatDouble(
                (double) visit_length_30s_60s / (double) session_count, 2);
        double visit_length_1m_3m_ratio = NumberUtils.formatDouble(
                (double) visit_length_1m_3m / (double) session_count, 2);
        double visit_length_3m_10m_ratio = NumberUtils.formatDouble(
                (double) visit_length_3m_10m / (double) session_count, 2);
        double visit_length_10m_30m_ratio = NumberUtils.formatDouble(
                (double) visit_length_10m_30m / (double) session_count, 2);
        double visit_length_30m_ratio = NumberUtils.formatDouble(
                (double) visit_length_30m / (double) session_count, 2);

        double step_length_1_3_ratio = NumberUtils.formatDouble(
                (double) step_length_1_3 / (double) session_count, 2);
        double step_length_4_6_ratio = NumberUtils.formatDouble(
                (double) step_length_4_6 / (double) session_count, 2);
        double step_length_7_9_ratio = NumberUtils.formatDouble(
                (double) step_length_7_9 / (double) session_count, 2);
        double step_length_10_30_ratio = NumberUtils.formatDouble(
                (double) step_length_10_30 / (double) session_count, 2);
        double step_length_30_60_ratio = NumberUtils.formatDouble(
                (double) step_length_30_60 / (double) session_count, 2);
        double step_length_60_ratio = NumberUtils.formatDouble(
                (double) step_length_60 / (double) session_count, 2);

        // 将统计结果封装为Domain对象
        SessionAggrStat sessionAggrStat = new SessionAggrStat();
        sessionAggrStat.setTask_id(task_id);
        sessionAggrStat.setSession_count(session_count);
        sessionAggrStat.setS1_s3(visit_length_1s_3s_ratio);
        sessionAggrStat.setS4_s6(visit_length_4s_6s_ratio);
        sessionAggrStat.setS7_s9(visit_length_7s_9s_ratio);
        sessionAggrStat.setS10_s30(visit_length_10s_30s_ratio);
        sessionAggrStat.setS30_s60(visit_length_30s_60s_ratio);
        sessionAggrStat.setM1_m3(visit_length_1m_3m_ratio);
        sessionAggrStat.setM3_m10(visit_length_3m_10m_ratio);
        sessionAggrStat.setM10_m30(visit_length_10m_30m_ratio);
        sessionAggrStat.setM30(visit_length_30m_ratio);
        sessionAggrStat.setV1_3(step_length_1_3_ratio);
        sessionAggrStat.setV4_6(step_length_4_6_ratio);
        sessionAggrStat.setV7_9(step_length_7_9_ratio);
        sessionAggrStat.setV10_30(step_length_10_30_ratio);
        sessionAggrStat.setV30_60(step_length_30_60_ratio);
        sessionAggrStat.setV60(step_length_60_ratio);
        // 调用对应的DAO插入统计结果
        ISessionAggrStatDAO sessionAggrStatDAO = DaoFactory.getSessionAggrStatDAOImpl();
        try {
            sessionAggrStatDAO.save(sessionAggrStat);
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }


    /**
     * 过滤session数据
     *
     * @param sessionid2AggrInfoRDD
     * @return
     */
    private static JavaPairRDD<String, String> filterSessionAndAggrStat(
            JavaPairRDD<String, String> sessionid2AggrInfoRDD,
            final JSONObject taskParam, final Accumulator<String> sessionAggrStatAccumulator) {
        // 为了使用我们后面的ValieUtils，所以，首先将所有的筛选参数拼接成一个连接串
        // 此外，这里其实大家不要觉得是多此一举
        // 其实我们是给后面的性能优化埋下了一个伏笔
        String startAge = ParamUtils.getParam(taskParam, Constants.PARAM_START_AGE);
        String endAge = ParamUtils.getParam(taskParam, Constants.PARAM_END_AGE);
        String professionals = ParamUtils.getParam(taskParam, Constants.PARAM_PROFESSIONALS);
        String cities = ParamUtils.getParam(taskParam, Constants.PARAM_CITIES);
        String sex = ParamUtils.getParam(taskParam, Constants.PARAM_SEX);
        String keywords = ParamUtils.getParam(taskParam, Constants.PARAM_KEYWORDS);
        String categoryIds = ParamUtils.getParam(taskParam, Constants.PARAM_CATEGORY_IDS);

        String _parameter = (startAge != null ? Constants.PARAM_START_AGE + "=" + startAge + "|" : "")
                + (endAge != null ? Constants.PARAM_END_AGE + "=" + endAge + "|" : "")
                + (professionals != null ? Constants.PARAM_PROFESSIONALS + "=" + professionals + "|" : "")
                + (cities != null ? Constants.PARAM_CITIES + "=" + cities + "|" : "")
                + (sex != null ? Constants.PARAM_SEX + "=" + sex + "|" : "")
                + (keywords != null ? Constants.PARAM_KEYWORDS + "=" + keywords + "|" : "")
                + (categoryIds != null ? Constants.PARAM_CATEGORY_IDS + "=" + categoryIds : "");

        if (_parameter.endsWith("\\|")) {
            _parameter = _parameter.substring(0, _parameter.length() - 1);
        }
        final String parameter = _parameter;

        // 根据筛选参数进行过滤

        /**
         * (5cb888caee744305b5f8f1df5bd4a4ff,sessionId=5cb888caee744305b5f8f1df5bd4a4ff|search_keyWord=新辣道鱼火锅,重庆辣子鸡,日本料理,蛋糕,国贸大厦|category_id=0,75,67,4,50,82|visit_length=2867|step_length=17|start_time=2018-04-06 15:01:09
         * |end_time=2018-04-06 15:48:56|age=46|professional=professional5|city=city96|sex=female)
         */
        JavaPairRDD<String, String> filteredSessionid2AggrInfoRDD = sessionid2AggrInfoRDD.filter(
                new Function<Tuple2<String, String>, Boolean>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public Boolean call(Tuple2<String, String> tuple) throws Exception {
                        // 首先，从tuple中，获取聚合数据
                        String aggrInfo = tuple._2;

                        // 接着，依次按照筛选条件进行过滤
                        // 按照年龄范围进行过滤（startAge、endAge）
                        if (!ValidUtils.between(aggrInfo, Constants.FIELD_AGE,
                                parameter, Constants.PARAM_START_AGE, Constants.PARAM_END_AGE)) {
                            return false;
                        }

                        // 按照职业范围进行过滤（professionals）
                        // 互联网,IT,软件
                        // 互联网
                        if (!ValidUtils.in(aggrInfo, Constants.FIELD_PROFESSIONAL,
                                parameter, Constants.PARAM_PROFESSIONALS)) {
                            return false;
                        }

                        // 按照城市范围进行过滤（cities）
                        // 北京,上海,广州,深圳
                        // 成都
                        if (!ValidUtils.in(aggrInfo, Constants.FIELD_CITY,
                                parameter, Constants.PARAM_CITIES)) {
                            return false;
                        }

                        // 按照性别进行过滤
                        // 男/女
                        // 男，女
                        if (!ValidUtils.equal(aggrInfo, Constants.FIELD_SEX,
                                parameter, Constants.PARAM_SEX)) {
                            return false;
                        }
                        // 按照搜索词进行过滤
                        // 我们的session可能搜索了 火锅,蛋糕,烧烤
                        // 我们的筛选条件可能是 火锅,串串香,iphone手机
                        // 那么，in这个校验方法，主要判定session搜索的词中，有任何一个，与筛选条件中
                        // 任何一个搜索词相当，即通过
                        if (!ValidUtils.in(aggrInfo, Constants.FIELD_SEARCH_KEYWORDS,
                                parameter, Constants.PARAM_KEYWORDS)) {
                            return false;
                        }
                        // 按照点击品类id进行过滤
                        if (!ValidUtils.in(aggrInfo, Constants.FIELD_CLICK_CATEGORY_IDS,
                                parameter, Constants.PARAM_CATEGORY_IDS)) {
                            return false;
                        }


                        // 如果经过了之前的多个过滤条件之后，程序能够走到这里
                        // 那么就说明，该session是通过了用户指定的筛选条件的，也就是需要保留的session
                        // 那么就要对session的访问时长和访问步长，进行统计，根据session对应的范围
                        // 进行相应的累加计数

                        // 主要走到这一步，那么就是需要计数的session
                        sessionAggrStatAccumulator.add(Constants.SESSION_COUNT);

                        // 计算出session的访问时长和访问步长的范围，并进行相应的累加
                        long visitLength = Long.valueOf(StringUtils.getFieldFromConcatString(
                                aggrInfo, "\\|", Constants.FIELD_VISIT_LENGTH));
                        System.out.println(aggrInfo);
                        long stepLength = Long.valueOf(StringUtils.getFieldFromConcatString(
                                aggrInfo, "\\|", Constants.FIELD_STEP_LENGTH));
                        calculateVisitLength(visitLength);
                        calculateStepLength(stepLength);
                        return true;
                    }

                    /**
                     * 计算访问时长范围
                     * @param visitLength
                     */
                    private void calculateVisitLength(long visitLength) {
                        if (visitLength >= 1 && visitLength <= 3) {
                            sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_1s_3s);
                        } else if (visitLength >= 4 && visitLength <= 6) {
                            sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_4s_6s);
                        } else if (visitLength >= 7 && visitLength <= 9) {
                            sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_7s_9s);
                        } else if (visitLength >= 10 && visitLength <= 30) {
                            sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_10s_30s);
                        } else if (visitLength > 30 && visitLength <= 60) {
                            sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_30s_60s);
                        } else if (visitLength > 60 && visitLength <= 180) {
                            sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_1m_3m);
                        } else if (visitLength > 180 && visitLength <= 600) {
                            sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_3m_10m);
                        } else if (visitLength > 600 && visitLength <= 1800) {
                            sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_10m_30m);
                        } else if (visitLength > 1800) {
                            sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_30m);
                        }
                    }

                    /**
                     * 计算访问步长范围
                     * @param stepLength
                     */
                    private void calculateStepLength(long stepLength) {
                        if (stepLength >= 1 && stepLength <= 3) {
                            sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_1_3);
                        } else if (stepLength >= 4 && stepLength <= 6) {
                            sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_4_6);
                        } else if (stepLength >= 7 && stepLength <= 9) {
                            sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_7_9);
                        } else if (stepLength >= 10 && stepLength <= 30) {
                            sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_10_30);
                        } else if (stepLength > 30 && stepLength <= 60) {
                            sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_30_60);
                        } else if (stepLength > 60) {
                            sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_60);
                        }
                    }

                });



        return filteredSessionid2AggrInfoRDD;
    }


    /**
     * 完成按session粒度到行为数据的聚合
     * 1. 完成用户行为数据的（userid,partInfo）的聚合
     * 2. 完成用户信息（userId，userInfo）的聚合
     * 3. 完成join的所有数据的聚合
     *
     * @param sqlContext sqlcontext
     * @param actionRDD  行为数据
     * @return
     */
    private static JavaPairRDD<String, String> aggregateBySession(SQLContext sqlContext, JavaRDD actionRDD, JavaPairRDD<String, Row> sessionid2ActionRDD) {

        // 现在actionRDD中的元素是Row，一个Row就是一行用户访问行为记录，比如一次点击或者搜索
        // 我们现在需要将这个Row映射成<sessionid,Row>的格式
        // 调用上面的映射函数
        // 对这个数据按照sessioId进行分组
        /**
         *(sessionid,list(actionInfo))------> sessionId+action集合
         */
        JavaPairRDD<String, Iterable<Row>> javaPairRDD = sessionid2ActionRDD.groupByKey();


        // 对每一个session分组进行聚合，将session中所有的搜索词和点击品类都聚合起来
        // 到此为止，获取的数据格式，如下：<userid,partAggrInfo(sessionid,searchKeywords,clickCategoryIds)>
        // 1. 完成行为数据的聚合
        /**
         *
         * （user,partInfo(sessionid,)）
         * (80,sessionId=376781d7506346bfa3ce986b155a5258|search_keyWord=重庆辣子鸡,新辣道鱼火锅,太古商场|
         * category_id=46,61,0,60,48|visit_length=3084|step_length=10|start_time=2018-04-06 02:01:20|end_time=2018-04-06 02:52:44)
         *
         */

        JavaPairRDD<Long, String> userid2PartAggrInfoRDD = javaPairRDD.mapToPair(new PairFunction<Tuple2<String, Iterable<Row>>, Long, String>() {
            @Override
            public Tuple2<Long, String> call(Tuple2<String, Iterable<Row>> tuple2) throws Exception {
                //1.获得userId
                Long userId = null;
                String sessionId = tuple2._1;
                Iterator<Row> iterabler = tuple2._2.iterator();
                StringBuffer searchKeywordsBuffer = new StringBuffer("");
                StringBuffer clickCategoryIdsBuffer = new StringBuffer("");
                // session的起始和结束时间
                Date startTime = null;
                Date endTime = null;
                // session的访问步长
                int stepLength = 0;

                // 遍历row，获取user_id
                while (iterabler.hasNext()) {
                    Row row = iterabler.next();
                    if (userId == null) {
                        userId = row.getLong(1);
                    }


                    // 获取search_keyword
                    // 获取click_category_id
                    String search_keyword = row.getString(5);
                    Long click_category_id = row.getLong(6);
                    // 实际上这里要对数据说明一下
                    // 并不是每一行访问行为都有searchKeyword何clickCategoryId两个字段的
                    // 其实，只有搜索行为，是有searchKeyword字段的
                    // 只有点击品类的行为，是有clickCategoryId字段的
                    // 所以，任何一行行为数据，都不可能两个字段都有，所以数据是可能出现null值的

                    // 我们决定是否将搜索词或点击品类id拼接到字符串中去
                    // 首先要满足：不能是null值
                    // 其次，之前的字符串中还没有搜索词或者点击品类id

                    if (StringUtils.isNotEmpty(search_keyword)) {
                        if (!searchKeywordsBuffer.toString().contains(search_keyword)) {
                            searchKeywordsBuffer.append(search_keyword + ",");
                        }
                    }
                    if (click_category_id != null) {
                        if (!clickCategoryIdsBuffer.toString().contains(
                                String.valueOf(click_category_id))) {
                            clickCategoryIdsBuffer.append(click_category_id + ",");
                        }
                    }
                    // 计算session开始和结束时间
                    Date actionTime = DateUtils.parseTime(row.getString(4));

                    if (actionTime != null) {

                        if (startTime == null) {
                            startTime = actionTime;
                        }
                        if (endTime == null) {
                            endTime = actionTime;
                        }

                        if (actionTime.before(startTime)) {
                            startTime = actionTime;
                        }
                        if (actionTime.after(endTime)) {
                            endTime = actionTime;
                        }
                        // 计算session访问步长
                        stepLength++;
                    }

                }

                // 计算session访问时长（秒）
                long visitLength = (endTime.getTime() - startTime.getTime()) / 1000;

                // 对搜索词和分类id进行去除空格
                String searchKeywords = StringUtils.trimComma(searchKeywordsBuffer.toString());
                String clickCategoryIds = StringUtils.trimComma(clickCategoryIdsBuffer.toString());

                // 拼接 partAggrInfo session的字符串

                String partAggrInfo = Constants.FIELD_SESSION_ID + "=" + sessionId + "|"
                        + (StringUtils.isNotEmpty(searchKeywords) ? Constants.FIELD_SEARCH_KEYWORDS + "=" + searchKeywords + "|" : "")
                        + (StringUtils.isNotEmpty(clickCategoryIds) ? Constants.FIELD_CLICK_CATEGORY_IDS + "=" + clickCategoryIds + "|" : "")
                        + Constants.FIELD_VISIT_LENGTH + "=" + visitLength + "|"
                        + Constants.FIELD_STEP_LENGTH + "=" + stepLength + "|"
                        + Constants.FIELD_START_TIME + "=" + DateUtils.formatTime(startTime) + "|"
                        + Constants.FIELD_END_TIME + "=" + DateUtils.formatTime(endTime);
                return new Tuple2<Long, String>(userId, partAggrInfo);
            }
        });



        //2. 完成用户数据的聚合

        // 查找用户数据

        String sql = "select * from user_info";

        DataFrame userData = sqlContext.sql(sql);
        JavaRDD<Row> userInfoRdd = userData.toJavaRDD();
        JavaPairRDD<Long, Row> userPair = userInfoRdd.mapToPair(new PairFunction<Row, Long, Row>() {
            @Override
            public Tuple2<Long, Row> call(Row row) throws Exception {
                return new Tuple2<Long, Row>(row.getLong(0), row);
            }
        });

        //3.对两个info进行聚合

        /**
         * (71,(sessionId=0276f343184d4b5eade395d8af3ad9cc|search_keyWord=温泉,日本料理,太古商场,新辣道鱼火锅|category_id=34,0,48|visit_length=2924|
         * step_length=12|start_time=2018-04-06 19:01:57|end_time=2018-04-06 19:50:41,[71,user71,name71,21,professional80,city13,female]))
         *
         */
        JavaPairRDD<Long, Tuple2<String, Row>> userFullInfoRdd = userid2PartAggrInfoRDD.join(userPair);

        // 对join起来的数据进行拼接，并且返回<sessionid,fullAggrInfo>格式的数据
        /**
         * (6a61393c1ca4467b8692b5c48da5b145,sessionId=6a61393c1ca4467b8692b5c48da5b145|search_keyWord=火锅,日本料理|category_id=3,0,35,97,41|visit_length=3363|step_length=11|start_time=2018-04-06 02:01:33|
         * end_time=2018-04-06 02:57:36|age=22|professional=professional53|city=city84|sex=male)
         */
        JavaPairRDD<String, String> sessionid2FullAggrInfoRDD = userFullInfoRdd.mapToPair(new PairFunction<Tuple2<Long, Tuple2<String, Row>>, String, String>() {
            @Override
            public Tuple2<String, String> call(Tuple2<Long, Tuple2<String, Row>> tupler) throws Exception {
                // 获取用户行为字符串
                String partInfo = tupler._2._1;
                //获取用户的信息row
                Row userRow = tupler._2._2;

                // 获取sessionId
                String sessionid = StringUtils.getFieldFromConcatString(
                        partInfo, "\\|", Constants.FIELD_SESSION_ID);

                int age = userRow.getInt(3);
                String professional = userRow.getString(4);
                String city = userRow.getString(5);
                String sex = userRow.getString(6);

                String fullAggrInfo = partInfo + "|"
                        + Constants.FIELD_AGE + "=" + age + "|"
                        + Constants.FIELD_PROFESSIONAL + "=" + professional + "|"
                        + Constants.FIELD_CITY + "=" + city + "|"
                        + Constants.FIELD_SEX + "=" + sex;
                return new Tuple2<String, String>(sessionid, fullAggrInfo);
            }
        });

        return sessionid2FullAggrInfoRDD;

    }


    /**
     * 获取sessionid2到访问行为数据的映射的RDD
     *
     * @param actionRDD
     * @return
     */
    private static JavaPairRDD<String, Row> getSessionid2ActionRDD(JavaRDD<Row> actionRDD) {
        // 进行seeionid的映射
        // 第一个参数是输入参数，是行为数据
        //第二，三个参数是输出参数，分别是sessionid和行为数据
        return actionRDD.mapToPair(new PairFunction<Row, String, Row>() {
            @Override
            public Tuple2<String, Row> call(Row row) throws Exception {
                return new Tuple2<String, Row>(row.getString(2), row);
            }
        });
    }


    /**
     * 根据时间范围，从user_visit_action查出数据
     *
     * @param sqlContext sqlContext
     * @param params     任务参数
     * @return 时间范围的rdd
     */
    private static JavaRDD<Row> getActionRDDByDateRange(SQLContext sqlContext, JSONObject params) {

        //参数中获取时间范围
        String startDate = ParamUtils.getParam(params, Constants.PARAM_START_DATE);
        String endDate = ParamUtils.getParam(params, Constants.PARAM_END_DATE);
        String sql =
                "select * "
                        + "from user_visit_action "
                        + "where date>='" + startDate + "' "
                        + "and date<='" + endDate + "'";
        DataFrame actionDf = sqlContext.sql(sql);
        return actionDf.javaRDD();
    }

    /**
     * 获取SQLContext
     * 如果是在本地测试环境的话，那么就生成SQLContext对象
     * 如果是在生产环境运行的话，那么就生成HiveContext对象
     *
     * @param sc SparkContext
     * @return SQLContext
     */
    private static SQLContext getSQLContext(SparkContext sc) {
        boolean local = ConfigurationManager.getBoolean(Constants.LOCAL_FLAGE);
        if (local) {
            return new SQLContext(sc);
        } else {
            return new HiveContext(sc);
        }
    }

    /**
     * 生成模拟数据（只有本地模式，才会去生成模拟数据）
     *
     * @param sc
     * @param sqlContext
     */
    private static void mockData(JavaSparkContext sc, SQLContext sqlContext) {
        boolean local = ConfigurationManager.getBoolean(Constants.LOCAL_FLAGE);
        if (local) {
            MockData.mock(sc, sqlContext);
        }
    }


    /**
     * 用于遍历rdd
     *
     * @param sessionid2AggrInfoRDD
     */

    private static void printRDD(JavaPairRDD<String, String> sessionid2AggrInfoRDD) {
        List<Tuple2<String, String>> sess = sessionid2AggrInfoRDD.take(10);

        for (Tuple2<String, String> tuple2 : sess) {
            System.out.println(tuple2._2);
        }
    }


}
